package com.study.flink.datastream

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

/**
  * Flink Split Select Demo
  *
  * @author stephen
  * @create 2019-05-25 14:55
  * @since 1.0.0
  */
object FlinkSplitSelectDemo {

  def main(args: Array[String]): Unit = {
    // 1、执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 2、指定数据源
    import org.apache.flink.api.scala._
    val dataStream = env.fromElements(("a", 1), ("b", 2), ("c", 3), ("a", 4), ("a", 5), ("b", 6))

    // 3、对数据进行操作
    // 切分数据集
    val splitedStream = dataStream.split(
      t =>
        t._2 % 2 == 0 match {
          case true => List("even")
          case false => List("odd")
        }
    )
    // 取出偶数
    val evenStream = splitedStream.select("even")
    // 取出奇数
    val oddStream = splitedStream.select("odd")
    // 取出多个数据集
    val allStream = splitedStream.select("even", "odd")

    // 4、指定数据输出位置
    evenStream.print()
    //oddStream.print()
    //allStream.print()

    // 5、触发任务执行
    env.execute("Flink Split Select Demo")
  }
}
